package com.yequan.mq.producer;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author yq
 * @date 2020/08/31
 */
public class ProducerQueue {

    public static void main(String[] args) throws Exception {

        //使用生产者组名称进行实例化
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        //指定nameserver的地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setRetryTimesWhenSendFailed(3);
//        producer.setDefaultTopicQueueNums(1);
        //启动生产者
        producer.start();
        AtomicInteger i = new AtomicInteger(0);
        for (; ; ) {
            //创建消息实例，指定topic、tag、消息内容
            //topic:只允许是^[%|a-zA-Z0-9_-]+$组成的字符串，不能是汉字
            Message msg = new Message("Hello", "TagA", ("Hello RocketMQ " + i.getAndIncrement()).getBytes(RemotingHelper.DEFAULT_CHARSET));
            //生产将消息发送给一个broker对象处理
            SendResult sendResult = producer.send(msg, 1000);
            System.out.println("发送对象信息为" + "   发送状态：" + sendResult.getSendStatus());
            Thread.sleep(1000);
//            Message msg2 = new Message("TopicTest2", "TagA", ("Hello RocketMQ222 " + i.getAndIncrement()).getBytes(RemotingHelper.DEFAULT_CHARSET));
//            SendResult sendResult2 = producer.send(msg2);
//            System.out.println("发送对象222信息为" + JSON.toJSONString(sendResult2));
            if (i.get() == 100) {
                break;
            }
        }
        //关闭生产者实例后此对象不再使用
        producer.shutdown();
    }

}
